from pyspark.sql import Window
from pyspark.sql.functions import *
from pyspark.sql.session import SparkSession
from pyspark.sql.types import StringType, IntegerType, StructType, StructField
import re

# 1、创建SparkSql执行环境
spark = SparkSession.builder.master("local").appName("sql").getOrCreate()

logs_df = (spark
           .read
           .format("csv")
           .option("sep", "\u0001")
           .schema("line STRING")
           .load("../../data/log")
           )

"""
自定义函数：如果使用spark sql中自带的函数处理不了的逻辑， 可以使用自定义函数处理
在自定义函数中可以使用python各种库处理数据
"""
# 1、定义函数解析log日志
def pas_log_fun(line: str):
    # 使用python正则表达式处理数据
    # 2025-06-07 14:46:01,564 INFO org.eclipse.jetty.util.log: Logging initialized @1993ms
    search = re.search(r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}) (\w{4,5}) ([\w\\.]+): (.+)", line)
    if search:
        time = search.group(1)
        log_type = search.group(2)
        clazz = search.group(3)
        comment = search.group(4)

        # 返回元组，按顺序匹配
        # return time, log_type, clazz, comment

        # 按照名称匹配
        return {
            "time": time,
            "log_type": log_type,
            "clazz": clazz,
            "comment": comment
        }

# 定义返回类型
return_type = StructType([
    StructField("time", StringType()),
    StructField("log_type", StringType()),
    StructField("clazz", StringType()),
    StructField("comment", StringType())
])

# 2、将python方法注册成spark sql的自定义函数
pas_log = udf(pas_log_fun, return_type)

# 3、使用自定义函数
log_pas_df = logs_df.select(pas_log("line").alias("log")).select("log.*").where(col("time").isNotNull())

log_pas_df.printSchema()
log_pas_df.show(truncate=False)

# 4、保存处理结果
log_pas_df.write.format("csv").option("sep", "\u0001").mode("overwrite").save("../../data/log_pas_log")



# 在sql中使用自定义函数
spark.udf.register("pas_log", pas_log_fun,return_type)
logs_df.createOrReplaceTempView("logs")
spark.sql("""
select log.* from (
select pas_log(line) log from logs
)
""").show()